package edu.seu.shuai.core;

import com.alibaba.fastjson.JSON;
import com.github.houbb.heaven.util.common.ArgUtil;
import com.github.houbb.id.core.util.IdHelper;
import com.github.houbb.log.integration.core.Log;
import com.github.houbb.log.integration.core.LogFactory;
import edu.seu.shuai.api.IMQProducer;
import edu.seu.shuai.constant.MethodType;
import edu.seu.shuai.constant.ProducerConst;
import edu.seu.shuai.constant.ProducerResponseCode;
import edu.seu.shuai.constant.SendStatus;
import edu.seu.shuai.dto.MQCommonRequest;
import edu.seu.shuai.dto.MQMessage;
import edu.seu.shuai.dto.SendResult;
import edu.seu.shuai.exception.MqException;
import edu.seu.shuai.handler.MQProducerHandler;
import edu.seu.shuai.response.MQCommonResp;
import edu.seu.shuai.response.MQCommonRespCode;
import edu.seu.shuai.rpc.RpcMessageDto;
import edu.seu.shuai.support.invoke.IInvokeService;
import edu.seu.shuai.utils.ChannelUtil;
import edu.seu.shuai.utils.DelimiterUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import edu.seu.shuai.support.invoke.impl.InvokeService;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

public class MQProducer extends Thread implements IMQProducer {

    private static final Log log = LogFactory.getLog(MQProducer.class);

    /**
     * 分组名称
     */
    private String groupName = ProducerConst.DEFAULT_GROUP_NAME;


    private int port = ProducerConst.DEFAULT_PORT;

    /**
     * 中间人地址
     */
    private String brokerAddress = "127.0.0.1:9999";

    /**
     * channel 信息
     *
     * @since 0.0.2
     */
    private ChannelFuture channelFuture;


    /**
     * 客户端处理 handler
     *
     * @since 0.0.2
     */
    private ChannelHandler channelHandler;

    /**
     * 调用管理服务
     *
     * @since 0.0.2
     */
    private final IInvokeService invokeService = new InvokeService();

    /**
     * 获取响应超时时间
     *
     * @since 0.0.2
     */
    private long respTimeoutMills = 5000;

    /**
     * 可用标识
     *
     * @since 0.0.2
     */
    private volatile boolean enableFlag = false;

    /**
     * 粘包处理分隔符
     *
     * @since 1.0.0
     */
    private String delimiter = DelimiterUtil.DELIMITER;

    public void run() {
        // 启动服务端
        log.info("MQ 生产者开始启动客户端 GROUP: {}, PORT: {}, brokerAddress: {}",
                groupName, port, brokerAddress);

        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {

            // channel handler
            this.initChannelHandler();

            Bootstrap bootstrap = new Bootstrap();
            this.channelFuture = bootstrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<Channel>(){
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            ch.pipeline()
                                    .addLast(new LoggingHandler(LogLevel.INFO))
                                    .addLast(new MQProducerHandler());
                        }
                    })
                    .connect("localhost", port)
                    .syncUninterruptibly();

            log.info("MQ 生产者启动客户端完成，监听端口：" + port);
            enableFlag = true;
            channelFuture.channel().closeFuture().syncUninterruptibly();
            log.info("MQ 生产者开始客户端已关闭");
        } catch (Exception e) {
            log.error("MQ 生产者启动遇到异常", e);
            throw new MqException(ProducerResponseCode.RPC_INIT_FAILED);
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    private void initChannelHandler() {
        log.info("initChannelHandler函数被执行了一次！");
        final ByteBuf delimiterBuf = DelimiterUtil.getByteBuf(delimiter);

        final MQProducerHandler mqProducerHandler = new MQProducerHandler();
        mqProducerHandler.setInvokeService(invokeService);

        // handler 实际上会被多次调用，如果不是 @Shareable，应该每次都重新创建。
        ChannelHandler handler = new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline()
                        .addLast(new DelimiterBasedFrameDecoder(DelimiterUtil.LENGTH, delimiterBuf))
                        .addLast(mqProducerHandler);
            }
        };
        this.channelHandler = handler;
    }
    @Override
    public SendResult send(MQMessage mqMessage) {
        String messageId = IdHelper.uuid32();
        mqMessage.setTraceId(messageId);
        mqMessage.setMethodType(MethodType.P_SEND_MESSAGE);
        MQCommonResp resp = callServer(mqMessage, MQCommonResp.class);
        if(MQCommonRespCode.SUCCESS.getCode().equals(resp.getRespCode())) {
            return SendResult.of(messageId, SendStatus.SUCCESS);
        }
        return SendResult.of(messageId, SendStatus.FAILED);
    }
    @Override
    public SendResult sendOneWay(MQMessage mqMessage) {
        String messageId = IdHelper.uuid32();
        mqMessage.setTraceId(messageId);
        mqMessage.setMethodType(MethodType.P_SEND_MESSAGE);
        this.callServer(mqMessage, null);
        return SendResult.of(messageId, SendStatus.SUCCESS);
    }

    private void paramCheck() {
        ArgUtil.notEmpty(groupName, "groupName");
        ArgUtil.notEmpty(brokerAddress, "brokerAddress");
    }

    /**
     * 调用服务端
     * @param commonReq 通用请求
     * @param respClass 类
     * @param <T> 泛型
     * @param <R> 结果
     * @return 结果
     * @since 1.0.0
     */
    public <T extends MQCommonRequest, R extends MQCommonResp> R callServer(T commonReq, Class<R> respClass) {
        final String traceId = commonReq.getTraceId();
        final long requestTime = System.currentTimeMillis();
        RpcMessageDto rpcMessageDto = new RpcMessageDto();
        rpcMessageDto.setTraceId(traceId);
        rpcMessageDto.setRequestTime(requestTime);
        rpcMessageDto.setJson(JSON.toJSONString(commonReq));
        rpcMessageDto.setMethodType(commonReq.getMethodType());
        rpcMessageDto.setRequest(true);
        // 添加调用服务
        invokeService.addRequest(traceId, respTimeoutMills);

        // 遍历 channel
        // 关闭当前线程，以获取对应的信息
        // 使用序列化的方式
        ByteBuf byteBuf = DelimiterUtil.getMessageDelimiterBuffer(rpcMessageDto);
        //负载均衡获取 channel
        Channel channel = channelFuture.channel();
        channel.writeAndFlush(byteBuf);
        String channelId = ChannelUtil.getChannelId(channel);

        log.debug("[Client] channelId {} 发送消息 {}", channelId, JSON.toJSON(rpcMessageDto));
        if (respClass == null) {
            log.debug("[Client] 当前消息为 one-way 消息，忽略响应");
            return null;
        } else {
            //channelHandler 中获取对应的响应
            RpcMessageDto messageDto = invokeService.getResponse(traceId);
            if (MQCommonRespCode.TIMEOUT.getCode().equals(messageDto.getRespCode())) {
                throw new MqException(MQCommonRespCode.TIMEOUT);
            }
            String respJson = messageDto.getJson();
            return JSON.parseObject(respJson, respClass);
        }
    }

    public boolean isEnableFlag() {
        return this.enableFlag;
    }
}
